package org.finesys.common.hazelcast.core.support;

import com.hazelcast.config.Config;
import com.hazelcast.config.NetworkConfig;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;

import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.StringJoiner;

@RequiredArgsConstructor
public class HazelcastInstanceFactory implements FactoryBean<HazelcastInstance>, DisposableBean {

    private final HazelcastProperties hazelcastProperties;
    private HazelcastInstance hazelcastInstance;

    public HazelcastInstance getInstance() {
        if (Objects.nonNull(hazelcastInstance)) {
            return hazelcastInstance;
        }

        // 配置实例名称
        Config config = new Config(hazelcastProperties.getInstanceName());
        // 配置集群名称
        config.setClusterName(hazelcastProperties.getClusterName());
        // 配置网络
        configNetwork(config.getNetworkConfig());
        // 创建实例
        hazelcastInstance = Hazelcast.newHazelcastInstance(config);

        return hazelcastInstance;
    }

    private void configNetwork(NetworkConfig networkConfig) {
        // 禁止多播，Hazelcast集群节点将无法自动加入到集群中
        networkConfig.getJoin().getMulticastConfig().setEnabled(false);
        // 启用TCP/IP协议，Hazelcast节点将通过TCP/IP协议加入到集群中。
        networkConfig.getJoin().getTcpIpConfig().setEnabled(true).setMembers(members()).setRequiredMember(null);
    }

    private List<String> members() {
        StringJoiner sj = new StringJoiner(",");
        hazelcastProperties.getMembers()
                .stream()
                .flatMap(m -> hazelcastProperties.getPort().stream().map(p -> m + ":" + p))
                .forEach(sj::add);
        return Arrays.asList(sj.toString().split(","));
    }

    @Override
    public HazelcastInstance getObject() throws Exception {
        return getInstance();
    }

    @Override
    public Class<?> getObjectType() {
        return HazelcastInstance.class;
    }

    @Override
    public void destroy() throws Exception {
        hazelcastInstance.shutdown();
    }
}
